﻿//
// Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail dot com)
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
// Official repository: https://github.com/boostorg/beast
//

//------------------------------------------------------------------------------
//
// Example: WebSocket server, asynchronous
//
//------------------------------------------------------------------------------


#include <algorithm>
#include <cstdlib>
#include <functional>

#include <memory>
#include <string>
#include <thread>

#include "websocket.h"

#define NO_DATA_RESTART

uint16_t port = 42002;
using namespace ws_;
using namespace std;

//void sendCheck(WsIoInterface* wp, std::shared_ptr< session> ss);
void sendCheck(ws_client* c);
void link_detect(tcp::acceptor *sock);
void calc_no_data_time(ws_client* c);
//------------------------------------------------------------------------------

//namespace beast = boost::beast;         // from <boost/beast.hpp>
//namespace http = beast::http;           // from <boost/beast/http.hpp>
//namespace websocket = beast::websocket; // from <boost/beast/websocket.hpp>
//namespace net = boost::asio;            // from <boost/asio.hpp>
//using tcp = boost::asio::ip::tcp;       // from <boost/asio/ip/tcp.hpp>

//WsIoInterface* wp = NULL;
ws_client* c = NULL;






inline
std::string
to_string(beast::flat_buffer const& buffer)
{
    return std::string(boost::asio::buffer_cast<char const*>(
        beast::buffers_front(buffer.data())),
        boost::asio::buffer_size(buffer.data()));
}

// Report a failure
void
fail(beast::error_code ec, char const* what)
{
    std::cerr << what << ": " << ec.message() << "\n";
}

void info(beast::error_code ec, char const* what)
{
    std::cerr << what << ": " << ec.message() << "\n";
}

// Echoes back all received WebSocket messages
// Get on the correct executor
void
session::run(WsIoInterface* p)
{
    // We need to be executing within a strand to perform async operations
    // on the I/O objects in this session. Although not strictly necessary
    // for single-threaded contexts, this example code is written to be
    // thread-safe by default.
    pp = p;
    net::dispatch(ws_.get_executor(),
        beast::bind_front_handler(
            &session::on_run,
            shared_from_this()));
}

// Start the asynchronous operation
void
session::on_run()
{
    // Set suggested timeout settings for the websocket
    ws_.set_option(
        websocket::stream_base::timeout::suggested(
            beast::role_type::server));

    // Set a decorator to change the Server of the handshake
    ws_.set_option(websocket::stream_base::decorator(
        [](websocket::response_type& res)
    {
        res.set(http::field::server,
            std::string(BOOST_BEAST_VERSION_STRING) +
            " websocket-server-async");
    }));

    ws_.read_message_max(1 << 26);//64MB
    //ws_.read_message_max(1<<20+4);//64MB 58839847
    ws_.binary(true);
    // Accept the websocket handshake
    ws_.async_accept(
        beast::bind_front_handler(
            &session::on_accept,
            shared_from_this()));
}

void
session::on_accept(beast::error_code ec)
{
    if (ec)
    {
        session_valid = false;
        return fail(ec, "accept");
    }
    info(ec, "wsAccept");
    //boost::asio::socket_base::send_buffer_size size_option(1024);
    //ws_.next_layer().socket().set_option(size_option);

    //boost::asio::socket_base::receive_buffer_size rx_size_option(1024);
    //ws_.next_layer().socket().set_option(rx_size_option);


    // Read a message
    do_RxTx();
}

void session::recvAsync()
{
    //rxBuffer_ = new beast::flat_buffer;
    ws_.async_read(
        rxBuffer_,
        beast::bind_front_handler(
            &session::on_read,
            shared_from_this()));
}

void session::sendAsync(std::string tmp)
{
    static std::string cc = tmp;
    //ostream(txBuffer_) << tmp;
    ws_.async_write(
        net::buffer(cc),
        beast::bind_front_handler(
            &session::on_write,
            shared_from_this()));

    //ws_.async_write(
    //	txBuffer_.data(),
    //	beast::bind_front_handler(
    //		&session::on_write,
    //		shared_from_this()));
    wbusy = true;

}
void session::do_RxTx()
{
    //start Tx ,Rx 
    recvAsync();
}

void
session::on_read(
    beast::error_code ec,
    std::size_t bytes_transferred)
{
    boost::ignore_unused(bytes_transferred);

    // This indicates that the session was closed
    if (ec == websocket::error::closed)
        return;

    if (ec)
    {
        session_valid = false;
        fail(ec, "read");
        return;
    }

    // Echo the messagesdxx
    //if (ws_.got_binary())
    //	ws_.binary(true);
    //else if(ws_.got_text())
    //	ws_.binary(true);
    //else
    //{}

    //############################
    t1 = time(NULL);
    if (!isFirst)
        isFirst = true;
    //############################
    size_t size = rxBuffer_.data().size();
    if (size == 0)
        recvAsync();
    char* d = (char*)rxBuffer_.data().data();
    std::string  s1;
    s1.assign(d, size);

    //wp->ProcessTaskPkg(to_string(rxBuffer_));
    pp->recvRawData(s1);
    rxBuffer_.consume(rxBuffer_.size());
    //recvcache.push_back(rxBuffer_);
    recvAsync();
}

void
session::on_write(
    beast::error_code ec,
    std::size_t bytes_transferred)
{
    boost::ignore_unused(bytes_transferred);

    if (ec)
    {
        session_valid = false;
        return fail(ec, "write");
    }

    // Clear the buffer
    //txBuffer_.consume(txBuffer_.size());
    wbusy = false;
    // Do another read
    //sendAsync();
}

//------------------------------------------------------------------------------

// Accepts incoming connections and launches the sessions
class listener : public std::enable_shared_from_this<listener>
{
    net::io_context& ioc_;
    tcp::acceptor acceptor_;
    std::vector<ws_client*>* ws_clientVector;
public:
    listener(
        net::io_context& ioc,
        std::vector<ws_client*>* vp,
        tcp::endpoint endpoint)
        : ioc_(ioc)
        , acceptor_(ioc)
    {
        ws_clientVector = vp;
        beast::error_code ec;

        // Open the acceptor
        acceptor_.open(endpoint.protocol(), ec);
        if (ec)
        {
            fail(ec, "open");
            return;
        }

        // Allow address reuse
        acceptor_.set_option(net::socket_base::reuse_address(true), ec);
        if (ec)
        {
            fail(ec, "set_option");
            return;
        }

        // Bind to the server address
        acceptor_.bind(endpoint, ec);
        if (ec)
        {
            fail(ec, "bind");
            return;
        }

        // Start listening for connections
        acceptor_.listen(
            net::socket_base::max_listen_connections, ec);
        if (ec)
        {
            fail(ec, "listen");
            return;
        }
    }

    // Start accepting incoming connections
    void
        run()
    {
        do_accept();
    }

private:
    void
        do_accept()
    {
        // The new connection gets its own strand
        acceptor_.async_accept(
            net::make_strand(ioc_),
            beast::bind_front_handler(
                &listener::on_accept,
                shared_from_this()));
    }
    //#define RAW_SENDER "192.168.10.28"
#define RAW_SENDER "127.0.0.1"
    void
        on_accept(beast::error_code ec, tcp::socket socket)
    {
        if (ec)
        {
            fail(ec, "accept");
        }
        else
        {
            // Create the session and run it
            //1.make client
            ws_client* c = NULL;
            cout << "\033[42m New client:\033[0m" << socket.remote_endpoint().address() << endl;
            cout << "port:" << socket.remote_endpoint().port() << endl;

            //socket will use for other way,so we get remote addr before it unvalid
            std::string raddr = socket.remote_endpoint().address().to_string();

            if (!memcmp(raddr.c_str(), RAW_SENDER, strlen(RAW_SENDER)))
            {
                if ((*ws_clientVector).size() && (*ws_clientVector).front())
                {
                    if ((*ws_clientVector).front()->session_sptr->session_valid)
                    {
                        //收到了来自同一端口的链接，且之前的链接有效，则判定为误操作链接，放弃这次链接
                        do_accept();
                        return;
                    }
                    else
                    {
                        //the old sesion is broken,so all of data in client 
                        //buffer is useless,free client and renew
                        delete (*ws_clientVector).front();
                        (*ws_clientVector).erase((*ws_clientVector).begin());
                    }
                }
                //instance class point use "new" operator,malloc or calloc 
                //does not support instance 嵌套类
                c = new ws_client;
                std::thread tc(link_detect, &acceptor_);
                tc.detach();
            }
            else
                c = new ws_client;

            //2.start session
            c->session_sptr = std::make_shared<session>(std::move(socket));
            c->session_sptr->run(&c->io);
            std::cout << "连接成功" << std::endl;
            c->session_sptr->session_valid = true;
            //3.start send process
            std::thread t(sendCheck, c);
            t.detach();

            //4.save client
            if (!memcmp(raddr.c_str(), RAW_SENDER, strlen(RAW_SENDER)))
            {
                (*ws_clientVector).insert((*ws_clientVector).begin(), c);
#ifdef NO_DATA_RESTART
                std::thread help2(calc_no_data_time, c);
                help2.detach();
#endif
            }
            else
                ws_clientVector->push_back(c);
            //pp.vector.push();

            //info(ec, "ACCEPT");

        }

        // Accept another connection
        do_accept();
    }
};

//------------------------------------------------------------------------------

//#############################
void calc_no_data_time(ws_client* c)
{
    std::shared_ptr< session> ss = c->session_sptr;
    //	time_t t1 = ss->t1;
    time_t t2;
    while (ss->session_valid)
    {
        if (ss->isFirst)
        {
            t2 = time(NULL);
            //printf("\033[45m start counting....\033[0m\n");
            if (t2 - (ss->t1) > 900)//no data over 15minus then restart app
            {
                printf("\033[42m no data over 15 minuts,acctuly :%lld \033[0m\n", t2 - ss->t1);
                exit(-1);
            }
            else
            {
                printf("\033[41m Checking data present, time gap = %lld \033[0m \n", t2 - ss->t1);
#ifdef WIN32
                Sleep(20000);
#else
                sleep(20);
#endif
            }
        }
        else
        {
#ifdef WIN32
            Sleep(6000);
#else
            sleep(60);
#endif
        }
    }
}
//##############################

void startListen(const char* addr, std::vector<ws_client*>* vp)
{
    auto const address = net::ip::make_address(addr);
    //auto const port = static_cast<unsigned short>(std::atoi("42002"));
    auto const threads = std::max<int>(1, std::atoi("1"));

    // The io_context is required for all I/O
    net::io_context ioc{ threads };

    // Create and launch a listening port
    std::make_shared<listener>(ioc, vp, tcp::endpoint{ address, port })->run();

    // Run the I/O service on the requested number of threads
    std::vector<std::thread> v;
    v.reserve(threads - 1);
    for (auto i = threads - 1; i > 0; --i)
        v.emplace_back(
            [&ioc]
    {
        ioc.run();
    });
    ioc.run();

    // return EXIT_SUCCESS;
}

//chect raw recieve handle is keep alive
void link_detect(tcp::acceptor *sock)
{
    //boost::asio::ip::tcp::socket socket(my_context);

    boost::asio::ip::tcp::socket::keep_alive option;
    while (1)
    {
        sock->get_option(option);
        bool is_set = option.value();
        cout << "socket alive:" << is_set << endl;
        std::this_thread::sleep_for(10s);
    }
}

void sendCheck(ws_client* c)
{
    //WsIoInterface *wp = &(c->io);
    std::shared_ptr< session> ss = c->session_sptr;
    beast::error_code ec;
    while (ss->session_valid)
    {
        if (ec == websocket::error::closed)
        {
            c->session_sptr->session_valid = false;
            break;
        }
        if (c->io.isResDataReady())
        {
            while (ss->wbusy);
            ss->sendAsync(c->io.getResData());
        }
    }

    cout << "ws sendRoutin exit!!\n";
}


void startServer(const char* addr, uint16_t userPort, std::vector<ws_client*>* p)
{
    //std::shared_ptr<std::thread> servHandle;
    //std::shared_ptr<std::thread> sendHandle;

    port = userPort;
    std::thread r(startListen, addr, p);

    //servHandle = std::make_shared<std::thread>(startListen(port));
    //sendHandle = std::make_shared<std::thread>(sendCheck);
    r.detach();

}

//void bindApi(WsIoInterface* p)
//{
//    wp = p;
//}